package com.stillalive.mq.mqserver.datacenter;

import com.stillalive.mq.common.BinaryTool;
import com.stillalive.mq.common.MqException;
import com.stillalive.mq.mqserver.core.MSGQueue;
import com.stillalive.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

public class MessageFileManager {
    // 验证版本，防止版本更新后代码与序列化的数据不匹配
    private static final long serialVersionUID = 1L;

    public void init(){
        // 此处暂不需要进行实际初始化工作
    }

    // 内部类表示队列的统计信息
    static public class Stat{
        public int totalCount;  // 总消息数量
        public int validCount;  // 有效消息数量
    }
    // 获取指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName){
        return "./data/" + queueName;
    }
    // 获取队列的消息数据文件路径
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/queue_data.txt";
    }
    // 获取队列的消息统计文件路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }
    private Stat readStat(String queueName){
        // 文本文件，直接用Scanner来读
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    private void writeStat(String queueName, Stat stat){
        // 使用PrintWriter来写文件
        // OutputStream默认打开文件会把原文件清空，相当于新的文件覆盖旧的，这里写的是stat对象，是需要覆盖之前的
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.println(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void createQueueFiles(String queueName) throws IOException{
        // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            boolean ok = baseDir.mkdir();
            if(!ok){
                throw new IOException("创建目录失败！baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();
            if(!ok){
                throw new IOException("创建文件失败！queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件设定初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    // 删除队列的目录和文件
    public void destroyQueueFiles(String queueName) throws IOException{
        // 先删除文件再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new  File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            throw new IOException("删除队列目录和文件失败！baseDir=" +  baseDir.getAbsolutePath());
        }
    }

    // 检查队列的目录和文件是否存在
    public boolean checkFileExists(String queueName) throws IOException{
        // 判断数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    // 把一个新的消息放在队列对应的文件中
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 检查队列对应文件是否存在
        if(!checkFileExists(queue.getName())){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在！queueName" + queue.getName());
        }
        // 2. 把Message对象进行序列化
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 线程安全，加锁
        synchronized (queue){
            // 3. 计算offsetBeg和 offsetEnd
            // 把新的Message数据写入到队列数据文件的末尾，此时Message对象的offsetBeg就是当前文件长度+4
            // offsetEnd就是当前文件长度+4+message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 4. 消息写入到数据文件，追加写
            try(OutputStream outputStream = new FileOutputStream(queueDataFile, true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    // 写入消息长度，4个字节，这里要保证四个字节，因此使用writeInt方法
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }
    // 删除某个Message，这里使用逻辑删除，isValid设置为0
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized(queue){
            // 这里根据offset就能找到对应文件的Message，因此使用RandomAccessFile随机读写随机访问
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")){
                // 从文件中读取对应的Message数据
                byte[] bufferSrc = new byte[(int)(message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 转回Message对象
                Message diskMessage = (Message)BinaryTool.fromBytes(bufferSrc);
                // 逻辑删除
                diskMessage.setIsValid((byte)0x0);
                // 重新写入，这里写的是操作后的diskMessage对象
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            if(stat.validCount > 0){
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    // 从文件中读取所有的消息内容加载到内存中，程序启动时进行调用
    // 该方法是在程序启动时调用的，此时服务器不会收到请求，因此不需要加锁，也就不需要使用queue对象，只需要queueName来找到对应队列文件即可
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws MqException, IOException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                long currentOffset = 0;
                while(true){
                    // 读取当前消息的长度
                    // 这里的readInt方法读到文件末尾时会抛出EOFException异常
                    int messageSize = dataInputStream.readInt();
                    // 按照这个长度读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(messageSize != actualSize){
                        // 如果不匹配，说明文件有问题，格式错乱了
                        throw new MqException("[MessageFileManager] 文件格式错误！queueName" + queueName);
                    }
                    // 读到的二进制数据反序列化回Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 判断消息对象是否为无效对象
                    if(message.getIsValid() != 0x1){
                        // 无效数据直接跳过
                        // 但是光标位置还是要改变的
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 有效数据，添加到链表中
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch(EOFException e){
                // 这里的异常处理处理正常的业务逻辑，当读取到文件末尾时readInt会抛出该异常，此时就执行到这里
                System.out.println("[MessageFileManager] 恢复Message数据完成！");
            }
        }
        return messages;
    }

    // 判断当前是否要针对该队列的消息数据文件进行GC
    public boolean checkGC(String queueName){
        Stat stat =  readStat(queueName);
        // 这里是自定义的GC条件，总消息数量超过2000且有效消息数量小于50%
        if(stat.validCount > 2000 && (double)stat.validCount / (double)stat.totalCount < 0.5){
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String  queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 执行真正的消息数据文件的垃圾回收操作
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized(queue){
            long gcBeg = System.currentTimeMillis();
            // 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                // 正常情况下这个文件不应该存在，如果存在，说明上次gc出现问题
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在！queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败！queueDataNewFile=" +  queueDataNewFile.getAbsolutePath());
            }
            // 从旧的文件读取出所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 把有效消息写入到新的文件
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 同样先写四个字节的消息长度
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 删除旧的数据文件，并且把新的文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败！queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！queueDataNewFile=" + queueDataNewFile.getAbsolutePath()
                + ", queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            // 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕！queueName= " + queue.getName() + "time=" + (gcEnd - gcBeg) + "ms");

        }
    }
}
